package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.bean.NoSink;
import com.atguigu.gmall.realtime.constant.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;

public class ClickhouseUtil {

   // 参数 sql :insert into books (id, title, authors, year) values (?, ?, ?, ?)
    public static <T> SinkFunction<T> getSink(String insertSql){

        SinkFunction<T> sink = JdbcSink.<T>sink(
                 insertSql,
                (statement, t ) -> {   // 把流中的元素字段与数据库字段做映射
                     //变量t中的所有字段值
                    Field[] fields = t.getClass().getDeclaredFields();
                    int skipNum=0;
                    for (int i = 0; i < fields.length; i++) {
                        Field field = fields[i];
                        field.setAccessible(true);
                        NoSink annotation = field.getAnnotation(NoSink.class);
                        if(annotation!=null){
                            skipNum++;
                            continue;
                        }
                        try {
                            statement.setObject(i+1-skipNum,field.get(t));
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(10000)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .build()
        );
        return sink;
    }
}
